package easier.kotlin

import kotlinx.coroutines.CoroutineName
import kotlinx.coroutines.CoroutineStart
import kotlinx.coroutines.DelicateCoroutinesApi
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.async
import kotlinx.coroutines.cancelAndJoin
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.newSingleThreadContext
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContext
import kotlinx.coroutines.withTimeout
import kotlin.system.measureTimeMillis

/**
 * 协程相关
 */
fun main() {
    test27()
}

fun test00() {
    println("start")/*runBlocking它用于阻塞线程并等待协程的执行完成*/
    runBlocking {
        /*启动一个协程处理*/
        launch {
            delay(1000)
            println("world")
        }
        println("hello")
    }
    println("end")

    //    start
    //    hello
    //    world
    //    end
}

fun test02() {
    println("start")/*runBlocking它用于阻塞线程并等待协程的执行完成*/
    runBlocking {
        /*协程作用域函数,等待其中的所有协程执行完成*/
        coroutineScope {
            launch {
                delay(1000)
                println("world1")
            }
            launch {
                delay(2000)
                println("world2")
            }
            println("hello")
        }/*等待协程处理完毕才执行*/
        println("hello out")
    }
    println("end")
    //    start
    //    hello
    //    world1
    //    world2
    //    hello out
    //    end
}

fun test03() {
    println("start")/*runBlocking它用于阻塞线程并等待协程的执行完成*/
    runBlocking {
        /*启动一个协程处理*/
        val launch = launch {
            delay(1000)
            println("world")
        }
        println("hello")/*需要等待协程处理完毕才能继续进行下面*/
        launch.join()
        println("done")
        //        hello
        //        world
        //        done
    }
    println("end")
}

/**
 * 协程是轻量级的,开始50000个协程也不会oom
 */
fun test04() {
    println("start")/*runBlocking它用于阻塞线程并等待协程的执行完成*/
    runBlocking {
        repeat(50_000) {
            launch {
                delay(2000)
                println("恰恰恰")
            }
        }
    }
    println("end")
}

/**
 *suspend 是一个关键字，用于标记挂起函数。挂起函数是一种特殊类型的函数，可以暂停执行而不阻塞线程
 * 只能在协程中调用,或者在其他suspend函数中
 *
 * runBlocking {
 *       println("start")
 *       test05()
 *       println("end")
 * }
 */
suspend fun test05() {
    delay(1000)
    println("world")
}

fun test06() {/*runBlocking它用于阻塞线程并等待协程的执行完成*/
    runBlocking {
        val job = launch {
            repeat(1000) { i ->
                println("sleeping $i ${System.currentTimeMillis()}")
                delay(500)
                println("wakeup  $i ${System.currentTimeMillis()}")
            }
        }
        delay(1100)
        println("tired ${System.currentTimeMillis()}")
        job.cancel()/*取消*/
        job.join()/*等待完成*/
        println("quit ${System.currentTimeMillis()}")

        /*每个任务500ms,等待了1300ms,已经完成两个,取消之后,正在进行的需要完成后才能结束,所以一共打印了3个*/
//        sleeping
//        sleeping
//        sleeping
//        tired
//        quit
    }
}

/**
 * 该实例中,因为没有响应cancel代码所以 job.cancelAndJoin()实际没有效果
 */
fun test07() {
    runBlocking {
        val job = launch(Dispatchers.Default) {
            var nextTime = System.currentTimeMillis()
            var i = 0
            while (i < 5) {/*不断运行,消耗cpu资源*/
                if (System.currentTimeMillis() >= nextTime) {/*每500ms打印一次,增加一下参数*/
                    println("sleeping ${i++}")
                    nextTime += 500
                }
            }
        }
        delay(1300)
        println(" tired")
        job.cancelAndJoin()
        println("finish")
    }
}

/**
 * repeat内部已经处理了job.cancelAndJoin(),所以可以取消,但是异常捕获之后,效果和test07一杨
 */
fun test08() {
    runBlocking {
        val job = launch {
            repeat(5) { i ->
                try {
                    println("sleeping $i")
                    delay(500)
                } catch (e: Exception) {
                    e.printStackTrace()
                }
            }
        }
        delay(1300)
        println(" tired")
        job.cancelAndJoin()
        println("finish")
    }
}

/**
 * 代码可以取消,
 */
fun test09() {
    runBlocking {
        val job = launch(Dispatchers.Default) {
            var nextTime = System.currentTimeMillis()
            var i = 0
            while (isActive && i < 5) {/*isActive,判断是否已经取消*//*不断运行,消耗cpu资源*/
                if (System.currentTimeMillis() >= nextTime) {/*每500ms打印一次,增加一下参数*/
                    println("sleeping ${i++}")
                    nextTime += 500
                }
            }
        }
        delay(1300)
        println("tired")
        job.cancelAndJoin()
        println("finish")

//        sleeping 0
//        sleeping 1
//        sleeping 2
//        tired
//        finish
    }
}

/**
 * 超时
 */
fun test10() {
    runBlocking {
        withTimeout(3300) {
            repeat(10) {
                println("sleeping")
                delay(500)
            }
        }
        println("finish")
    }
}

var acquire = 0

class Resource {
    init {
        acquire++
    }

    fun close() {
        acquire--
    }
}

/**
 * 并不总是打印0,因为在这个协程的竞争中有的协程超时了,所以没有close掉
 */
fun test11() {
    runBlocking {
        repeat(10_000) {
            launch {
                try {
                    val resource = withTimeout(60) {
                        delay(50)
                        Resource()
                    }
                    resource.close()
                } catch (e: Exception) {
                    println(e)
                }
            }
        }
    }
    println("$acquire")
}

/**
 * finally中
 */
fun test12() {
    runBlocking {
        repeat(10_000) {
            launch {
                var resource: Resource? = null
                try {
                    withTimeout(60) {
                        delay(50)
                        resource = Resource()/*超时异常可能发生在代码块的任何地方,即使是return的地方,所以放在外边是可能不会赋值的*/
                    }
                } finally {
                    resource?.close()
                }
            }
        }
    }
    println("$acquire")
}

suspend fun doSomeThingOne(): Int {
    delay(1000)
    return 13
}

suspend fun doSomeThingTwo(): Int {
    delay(2000)
    return 13
}

fun test13() {
    runBlocking {
        val time = measureTimeMillis {
            /*以下suspend方法顺序调用,共耗时3000ms*/
            val one = doSomeThingTwo()
            val two = doSomeThingOne()
            println("the result is $one + $two = ${one + two}")
        }
        println("cast is $time")
    }
}

/**
 * 异步调用suspend方法
 */
fun test14() {
    runBlocking {
        val time = measureTimeMillis {
            /*异步同时进行,消耗时间2000*/
            val one = async { doSomeThingOne() }
            val two = async { doSomeThingTwo() }
            println("the result is ${one.await()} + ${two.await()} = ${one.await() + two.await()}")
        }
        println("cast is $time")
    }
}

/**
 * 惰性启动
 */
fun test15() {
    runBlocking {
        val time = measureTimeMillis {
            /*异步同时进行,消耗时间2000*/
            val one = async(start = CoroutineStart.LAZY) { doSomeThingOne() }
            val two = async(start = CoroutineStart.LAZY) { doSomeThingTwo() }
            one.start()
            two.start()
            println("the result is ${one.await()} + ${two.await()} = ${one.await() + two.await()}")
        }
        println("cast is $time")
    }
}

/**
 * @OptIn(DelicateCoroutinesApi::class) 注解的作用是标记使用了不稳定或不推荐使用的 Kotlin 协程 API，并通过编译器的警告帮助开发者保证代码的稳定性和可靠性
 */
@OptIn(DelicateCoroutinesApi::class)
fun doSomeThingOneAsync() = GlobalScope.async {
    doSomeThingOne()
}

@OptIn(DelicateCoroutinesApi::class)
fun doSomeThingTwoAsync() = GlobalScope.async {
    doSomeThingTwo()
}

/**
 * GlobalScope.async可以编写任意的函数
 */
fun test16() {
    val time = measureTimeMillis {
        val one = doSomeThingOneAsync()
        val two = doSomeThingTwoAsync()
        runBlocking {
            println("the result is ${one.await()} + ${two.await()} = ${one.await() + two.await()}")
        }
    }
    println("cast is $time")
}

/**
 *
 */
fun test17() {

    runBlocking {
        try {
            failedCoroutineSum()
        } catch (e: Exception) {
            println("catch a exception $e")
        }
    }
}

/**
 * coroutineScope创建一个新的协程作用域
 */
suspend fun failedCoroutineSum(): Int = coroutineScope {
    val one = async<Int> {
        try {
            delay(Long.MAX_VALUE)
            13
        } finally {
            println("one is cancelled")
        }
    }

    val two = async<Int> {
        delay(2000)
        println("two throw a exception")
        throw ArithmeticException()
    }

    one.await() + two.await()
}

fun test18() {
    runBlocking {
        launch {
            println("launch默认 run ${Thread.currentThread().name}")
        }

        launch(Dispatchers.Default) {
            println("launch(Dispatchers.Default) run ${Thread.currentThread().name}")
        }

        launch(Dispatchers.Unconfined) {
            println("launch(Dispatchers.Unconfined) run ${Thread.currentThread().name}")
        }

        launch(newSingleThreadContext("myOwnThread")) {
            println("launch(newSingleThreadContext) run ${Thread.currentThread().name}")
        }
    }
}

fun test19() {
    runBlocking {
        launch(Dispatchers.Unconfined) {
            println("start work in Unconfined ${Thread.currentThread().name}")
            delay(2000)/*delay之后线程切换到了 kotlinx.coroutines.DefaultExecutor*/
            println("after work in Unconfined ${Thread.currentThread().name}")
        }

        launch {
            println("start work in ${Thread.currentThread().name}")
            delay(2000)
            println("after work in ${Thread.currentThread().name}")
        }
    }
}

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg ${System.currentTimeMillis()}")

fun test20() {
    newSingleThreadContext("ctx0").use { ctx0 ->
        newSingleThreadContext("ctx1").use { ctx1 ->
            runBlocking(ctx0) {
                log("start in ctx0")/*withContext 函数本身是一个挂起函数，它会暂停当前协程的执行，然后切换到指定的上下文，并在内部代码块执行完毕后恢复该协程的执行。*/
                withContext(ctx1) {
                    log("work start in ctx1")
                    delay(1000)
                    log("work finish in ctx1")
                }
                log("back to ctx0")
            }
        }
    }
}

fun test21() {
    runBlocking {
        println("my job is $coroutineContext[Job]")
    }
}

fun test22() {
    runBlocking {
        val job = launch {
            /*repeat 依次执行代码*/
            repeat(3) { i ->
                delay((i + 1) * 200L)
                log("coroutine is done")
            }
            log("start task")
        }
        job.join()
        log("all is done")
    }
}

fun test23() {
    runBlocking {
        log("start in main thread")
        val v1 = async(CoroutineName("V1")) {
            delay(500)
            log("compute V1")
            256
        }

        val v2 = async(CoroutineName("V2")) {
            delay(1000)
            log("compute V2")
            6
        }

        log("result is ${v1.await() / v2.await()}")
    }
}

fun simple(): List<Int> = listOf(1, 2, 3)

fun test24() {
    simple().forEach { item ->
        println("item is $item")
    }
}

fun simple02(): Sequence<Int> = sequence {
    for (i in 1..3) {
        Thread.sleep(1000)
        yield(i)
    }
}

fun test25() {
    simple02().forEach { i ->
        log(i.toString())
    }
}

suspend fun simple03(): List<Int> {
    delay(1000)
    return listOf(1, 2, 3)
}

fun test26() {
    runBlocking {
        println("start")
        simple03().forEach { i ->
            log(i.toString())
        }
        println("end")
    }
}

fun simple04(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(100)
        emit(i)
    }
}

fun test27() {
    runBlocking {
        launch {
            for (i in 1..3) {
                delay(100)
                println("this is $i")
            }
        }
        simple04().collect { item ->
            println(item.toString())
        }
    }
}